package com.taobao.analyse.map;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;

import org.apache.hadoop.io.DataInputBuffer;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;
import org.springframework.util.CollectionUtils;

import com.taobao.analyse.constants.SplitConstants;
import com.taobao.analyse.utils.LCDataParser;
import com.taobao.analyse.utils.SimpleDateFormatUtils;
import com.taobao.cmp.citadel.message.MessageType;
import com.taobao.cmp.citadel.message.PropConstants;
import com.taobao.cmp.citadel.message.SkuUtil;
import com.taobao.cmp.lib.sst.Document;
import com.taobao.cmp.lib.sst.mapred.writable.SSTWritable;
import com.taobao.item.constant.ItemSkuUpdateChangedField;
import com.taobao.item.domain.ItemSkuDO;

/**
 * 
 * @author tieyi.qlr
 */
public class ItemPriceChangedMapper extends MapReduceBase implements Mapper<NullWritable, SSTWritable, LongWritable, Text> {
	private DataInputBuffer	input	= new DataInputBuffer();
	
	@Override
	public void configure(JobConf jobConf) {
	}

	@Override
	public void map(NullWritable key, SSTWritable value, OutputCollector<LongWritable, Text> output, Reporter reporter)
			throws IOException {
		try {
			Document doc = value.getDoc();
			Integer level = (Integer) doc.getField("level").getValue();

			//
			if (level <= 0) {
				reporter.incrCounter("LCParser", "Level0", 1);
				return;
			}

			byte[] payload = (byte[]) doc.getField("payload").getValue();
			input.reset(payload, payload.length);
			Map<String, Object> val = LCDataParser.read(input, reporter);
			
			Object aid = val.get(PropConstants.ITEM_ID);
			//商品id为空，搞毛啊
			if(aid == null){
				return;
			}
			long itemId = Long.parseLong(String.valueOf(aid));
			//消息类型
			String operateType = String.valueOf(val.get(PropConstants.MESSAGE_TYPE));
			
			
			//先获取商品上的一口价价格是否有变更
			Object auctionPrice = val.get(PropConstants.PRICE);
			//获得sku列表的变更记录
			List<ItemSkuDO> skus = SkuUtil.parseBytes((byte[]) val
					.get(PropConstants.SKU));
			
			//暂时使用。。。
			//output.collect(new LongWritable(itemId), new Text(operateType));
			
			if (operateType.equals(MessageType.IC_SELLER_EDIT)) {
				//此块只记录卖家编辑的数据
				//操作时间
				Object operat_time = val.get("time");
				
				//如果商品上价格和sku上的价格都没有变化，则略过些条数据
				if(auctionPrice == null && CollectionUtils.isEmpty(skus)){
					output.collect(new LongWritable(itemId), new Text(buildUpdateCountInfo(operat_time,false)));
					return;
				}else{
					output.collect(new LongWritable(itemId), new Text(buildUpdateCountInfo(operat_time,true)));
				}
				
				StringBuilder sb = new StringBuilder();
				buildUpdateRecord(sb,operat_time,auctionPrice,skus);
				//输出结果吧
				output.collect(new LongWritable(itemId), new Text(sb.toString()));
			}else{
				return;
			}
			
		} catch (Exception e) {
			reporter.incrCounter("LCParser", "MapperExe", 1);
		}
	}

	/**
	 * 对于新发的宝贝构建源数据，用于后面的统计
	 * source@price:1234;sku:0_0,1_100,2_200
	 * @param sb
	 * @param auctionPrice
	 * @param skus
	 */
	private void buildNewPublishItemRecord(StringBuilder sb,
			Object auctionPrice, List<ItemSkuDO> skus) {
		sb.append("source").append(SplitConstants.SPLIT_AT);
		buildAuctionPrice(sb,auctionPrice);
		sb.append(SplitConstants.SPLIT_SEMCOLON);
		buiidSkus(sb,skus);
	}

	/**
	 * 用于累加细加计数，如果编辑价格则记录中会置1，格式如：rate_2012-05-01:0
	 * 0表示未编辑价格，1表示编辑了价格字段
	 * @param operat_time
	 * @param editPrice 是否编辑价格
	 * @return
	 */
	private String buildUpdateCountInfo(Object operat_time, boolean editPrice) {
		StringBuilder sb = new StringBuilder();
		Date updateTime = new Date(Long.valueOf(String.valueOf(operat_time)));
		String timeStr = SimpleDateFormatUtils.converDateToStr("yyyy-MM-dd", updateTime);
		sb.append("rate")
		.append(SplitConstants.SPLIT_UNDER_BAR)
		.append(timeStr)
		.append(SplitConstants.SPLIT_COLON)
		.append(editPrice?1:0);
		return sb.toString();
	}

	/**
	 * 生成编辑价格的记录，用于二次分析
	 * 封装格式：2012-03-32 12:32:32~price:1234;sku:0_0,1_100,2_200,
	 * @param sb
	 * @param operat_time
	 * @param auctionPrice
	 * @param skus
	 */
	public void buildUpdateRecord(StringBuilder sb, Object operat_time,
			Object auctionPrice, List<ItemSkuDO> skus) {
		
		buildOperationTime(sb,operat_time);
		sb.append(SplitConstants.SPLIT_WAVY);
		
		buildAuctionPrice(sb,auctionPrice);
		sb.append(SplitConstants.SPLIT_SEMCOLON);
		
		buiidSkus(sb,skus);
	}

	/**
	 * 生成sku变更的组装记录
	 * 如果没有一个sku的价格进行变更，则输出：sku:null
	 * 若有，则输出：sku:skuId1_skuPrice1,skuId2_skuPrice2,
	 * @param sb
	 * @param skus
	 */
	private void buiidSkus(StringBuilder sb, List<ItemSkuDO> skus) {
		//生成sku:
		sb.append(PropConstants.SKU).append(SplitConstants.SPLIT_COLON);
		if(CollectionUtils.isEmpty(skus)){
			sb.append("null");
		}else{
			boolean hasOneSkuChangePrice = false;
			for(ItemSkuDO sku : skus){
				if(sku.isChangedSomeField(ItemSkuUpdateChangedField.PRICE)){
					hasOneSkuChangePrice = true;
					sb.append(sku.getSkuId()).append(SplitConstants.SPLIT_UNDER_BAR).append(sku.getPrice());
					sb.append(SplitConstants.SPLIT_COMMA);
				}
			}
			if(!hasOneSkuChangePrice){
				sb.append("null");
			}
		}
	}

	/**
	 * 增加aa表上价格的编辑信息
	 * 如果此字段没有变更，则生成记录：price:null
	 * 如果字段有变更，则生成记录：price:auctionPrice的值
	 * @param sb
	 * @param auctionPrice
	 */
	private void buildAuctionPrice(StringBuilder sb, Object auctionPrice) {
		sb.append(PropConstants.PRICE).append(SplitConstants.SPLIT_COLON);
		if(auctionPrice == null){
			sb.append("null");
		}else{
			sb.append(auctionPrice);
		}
	}

	/**
	 * 生成时间变更的记录(operationTime-)，如2012-03-32 12:32:32
	 * @param sb
	 * @param operat_time
	 */
	private void buildOperationTime(StringBuilder sb, Object operat_time) {
		sb.append(operat_time);
	}
	
	public static void main(String[] args) {
		ItemPriceChangedMapper mapper = new ItemPriceChangedMapper();
		StringBuilder sb = new StringBuilder();
		String operat_time = "2012-03-32 12:32:32";
		Object auctionPrice = null;
		List<ItemSkuDO> skus = new ArrayList<ItemSkuDO>();
		for(int i=0;i<3;i++){
			ItemSkuDO sku = new ItemSkuDO();
			sku.setSkuId(i);
			sku.setPrice(i*100);
			sku.addChangedField(ItemSkuUpdateChangedField.PRICE);
			skus.add(sku);
		}
		
		if(auctionPrice == null && CollectionUtils.isEmpty(skus)){
			System.out.println("all is null! fuck");
		}
		
		mapper.buildUpdateRecord(sb, operat_time, auctionPrice, skus);
		System.out.println(sb.toString());
	}

}
